package com.atguigu.kafka.flink.producer;

import com.alibaba.fastjson2.JSONObject;
import com.atguigu.kafka.flink.model.Person;
import javafx.util.Duration;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.time.LocalDate;
import java.time.LocalDateTime;
import java.time.LocalTime;
import java.util.Properties;

public class MyKafkaProdecer {
    public static void main(String[] args) {

        //0.配置
        Properties properties = new Properties();

        //连接集群
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"10.254.131.147:9092");

        //指定对应的key和value
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        //1.创建kafka生产者对象
        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(properties);
        //2.发送数据
        for (int i = 0; i <= 30; i++) {
            //随机生成50个人，已经他们月收入
            Person person=new Person("cq",18,"男", i*100, LocalDateTime.of(LocalDate.now(), LocalTime.MIN).plusSeconds(2*i));
            String jsonString = JSONObject.toJSONString(person);
            kafkaProducer.send(new ProducerRecord<>("flink_test",jsonString));
        }
        for (int i = 0; i <= 30; i++) {
            //随机生成50个人，已经他们月收入
            Person person=new Person("cq2",18,"男", i*100,LocalDateTime.of(LocalDate.now(), LocalTime.MIN).plusSeconds(2*i));
            String jsonString = JSONObject.toJSONString(person);
            kafkaProducer.send(new ProducerRecord<>("flink_test",jsonString));
        }
        for (int i = 0; i <= 30; i++) {
            //随机生成50个人，已经他们月收入
            Person person=new Person("cq3",18,"男", i*100,LocalDateTime.of(LocalDate.now(), LocalTime.MIN).plusSeconds(2*i));
            String jsonString = JSONObject.toJSONString(person);
            kafkaProducer.send(new ProducerRecord<>("flink_test",jsonString));
        }
        for (int i = 0; i <= 30; i++) {
            //随机生成50个人，已经他们月收入
            Person person=new Person("cq4",18,"男", i*100,LocalDateTime.of(LocalDate.now(), LocalTime.MIN).plusSeconds(2*i));
            String jsonString = JSONObject.toJSONString(person);
            kafkaProducer.send(new ProducerRecord<>("flink_test",jsonString));
        }
        for (int i = 0; i <= 30; i++) {
            //随机生成50个人，已经他们月收入
            Person person=new Person("cq5",18,"男", i*100,LocalDateTime.of(LocalDate.now(), LocalTime.MIN).plusSeconds(2*i));
            String jsonString = JSONObject.toJSONString(person);
            kafkaProducer.send(new ProducerRecord<>("flink_test",jsonString));
        }

        //3.关闭资源
        kafkaProducer.close();
    }
}
